package nettyRPC.client;

import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import nettyRPC.proxy.RpcClientHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @program: NettyRpc-master
 * @description
 * @author: tkk fendoukaoziji
 * @create: 2019-08-12 18:36
 **/
public class ConnectManger {
    private static final Logger logger=LoggerFactory.getLogger(ConnectManger.class);
    private static volatile ConnectManger connectManger;
    private volatile boolean isRuning = true;
    private EventLoopGroup eventLoopGroup=new NioEventLoopGroup(4);
    private static ThreadPoolExecutor threadPoolExecutor=new ThreadPoolExecutor(
            16,16,600L,TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(65536)
    );
    private CopyOnWriteArrayList<RpcClientHandler> connectedHandlers =new CopyOnWriteArrayList<>();

    private Map<InetSocketAddress ,RpcClientHandler> connectedServerNodes =new ConcurrentHashMap<>();

    private ReentrantLock lock = new ReentrantLock();
    private Condition connected = lock.newCondition();
    private long connectTimeoutMillis = 6000;
    private AtomicInteger roundRobin = new AtomicInteger(0);

    private ConnectManger(){

    }

    public static ConnectManger getInstance(){
        if(connectManger==null){
            synchronized (ConnectManger.class){
                if(connectManger==null){
                    connectManger=new ConnectManger();
                }
            }
        }
        return connectManger;
    }

    private boolean waitingForHandler() throws InterruptedException {
        lock.lock();
        try {
            return connected.await(this.connectTimeoutMillis,TimeUnit.MILLISECONDS);
        } finally {
            lock.unlock();
        }
    }




    public RpcClientHandler chooseHandler(){
        int size = connectedHandlers.size();
        while (isRuning&& size<=0){
            try {
                boolean available = waitingForHandler();
                if(available){
                    size= connectedHandlers.size();
                }
            } catch (InterruptedException e) {
                logger.error("Waiting for available node is interrupted! ", e);
                throw new RuntimeException("Can't connect any servers!", e);
            }

        }
        int index = (roundRobin.getAndAdd(1) + size) % size;
        return connectedHandlers.get(index);
    }

    public void updateConnectedServer(List<String>allServerAddress){
        if(allServerAddress!=null){
            if(allServerAddress.size()>0){
                //update local sesrverNode cache
                HashSet<InetSocketAddress> newAllServerNodeSet=new HashSet<InetSocketAddress>();
                for (int i = 0; i < allServerAddress.size(); i++) {
                    String[] array = allServerAddress.get(i).split(":");
                    if(array.length==2){
                        String host=array[0];
                        int port=Integer.parseInt(array[1]);
                        final InetSocketAddress remotePeer=new InetSocketAddress(host,port);

                        newAllServerNodeSet.add(remotePeer);
                    }
                }

                //add new server node
                for(final InetSocketAddress serverNodeAddress:newAllServerNodeSet){
                    if(!connectedServerNodes.keySet().contains(serverNodeAddress)){
                        connectServerNode(serverNodeAddress);
                    }
                }
                //close and remove invalid server node
                for (int i = 0; i < connectedHandlers.size(); i++) {
                    RpcClientHandler connectedServerHandler= connectedHandlers.get(i);
                    SocketAddress remotePeer = connectedServerHandler.getRemotePeer();
                    if(!newAllServerNodeSet.contains(remotePeer)){
                        logger.info("Remove invalid server node " + remotePeer);
                        RpcClientHandler handler= connectedServerNodes.get(remotePeer);
                        if(handler!=null){
                            handler.close();
                        }
                        connectedServerNodes.remove(remotePeer);
                        connectedHandlers.remove(connectedServerHandler);
                    }
                }

            }else{ //All server nodes are down
                logger.error("No avaliable server node All server nodes are down!!!");
                for (final RpcClientHandler conServerHandler: connectedHandlers) {
                    SocketAddress remoteUser = conServerHandler.getRemotePeer();
                    RpcClientHandler handler = connectedServerNodes.get(remoteUser);
                    handler.close();
                    connectedServerNodes.remove(conServerHandler);
                }
                connectedHandlers.clear();
            }
        }
    }

    private void connectServerNode(InetSocketAddress serverNodeAddress) {

    }

    public void  stop(){

    }
}











